package com.shujia.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;

public class Demo8Time {
    Connection connection;

    /**
     * 创建链接
     */
    @Before
    public void init() {
        //1、创建hbase链接
        //创建hadoop配置对象
        Configuration configuration = HBaseConfiguration.create();
        //指定链接参数
        //指定zk地址，在zk中保存了hbase元数据
        configuration.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");
        //建立链接
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Test
    public void puts() throws Exception {
        //create 'dianxin_time','info'
        Table dianxinSplit = connection.getTable(TableName.valueOf("dianxin_time"));

        ArrayList<Put> puts = new ArrayList<>();

        BufferedReader bufferedReader = new BufferedReader(new FileReader("../data/DIANXIN.csv"));

        String line;
        while ((line = bufferedReader.readLine()) != null) {
            String[] split = line.split(",");
            String mdn = split[0];
            String startTime = split[1];
            String lon = split[4];
            String lat = split[5];

            //用大数减去小数，将时间升序转换成降序
            long end = 40000000000000L - Long.parseLong(startTime);

            String rowkey = mdn + "_" + end;

            Put put = new Put(Bytes.toBytes(rowkey));

            //增加列
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("lon"), Bytes.toBytes(lon));
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("lat"), Bytes.toBytes(lat));

            puts.add(put);
        }


        dianxinSplit.put(puts);
    }

}
